package com.dddframework.mq.config;

import cn.hutool.core.map.MapUtil;
import com.dddframework.core.context.BaseContext;
import com.dddframework.core.context.SpringContext;
import com.dddframework.core.contract.constant.ContextConstants;
import com.dddframework.mq.kafka.core.BaseMQConsumer;
import com.dddframework.mq.kafka.impl.BaseKafkaConsumer;
import com.dddframework.mq.redis.IMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.hash.ObjectHashMapper;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.List;
import java.util.Objects;

@Configuration
@EnableConfigurationProperties(BaseMQProperties.class)
@Slf4j
public class BaseMQConfig {

    @Bean
    public BaseMQConsumer baseMQConsumer() {
        return new BaseMQConsumer();
    }

    @Bean
    @ConditionalOnProperty(name = "spring.kafka.consumer.auto-startup", havingValue = "true")
    public BaseKafkaConsumer baseKafkaConsumer() {
        return new BaseKafkaConsumer();
    }

    @Bean(initMethod = "start", destroyMethod = "stop")
    @Primary
    public StreamMessageListenerContainer<String, ObjectRecord<String, String>> container(List<IMessageListener> iMessageListenerList) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                // 一次性最多消费消息
                .batchSize(10)
                // 使用默认的
                //.executor()
                //.keySerializer(new StringRedisSerializer())
                //.hashKeySerializer(new StringRedisSerializer())
                .hashValueSerializer(new StringRedisSerializer())
                // Stream 中没有消息时，阻塞多长时间，必须要比 `spring.redis.timeout` 小
                .pollTimeout(Duration.ofSeconds(1))
                // ObjectRecord 时，将 对象的 filed 和 value 转换成一个 Map 比方：将Book对象转换成map
                .objectMapper(new ObjectHashMapper())
                // 异常处理
                // .errorHandler(new CustomErrorHandler())
                // 将发送到Stream中的Record转换成ObjectRecord，转换成具体的类型
                .targetType(String.class)
                .build();
        RedisTemplate redisTemplate = BaseContext.get(ContextConstants.REDIS_TEMPLATE);
        StreamMessageListenerContainer<String, ObjectRecord<String, String>> container = StreamMessageListenerContainer.create(redisTemplate.getConnectionFactory(), options);
        for (IMessageListener listener : iMessageListenerList) {
            createGroup(redisTemplate, listener.queue(), listener.group());
            StreamMessageListenerContainer.StreamReadRequest streamReadRequest = StreamMessageListenerContainer.StreamReadRequest.builder(StreamOffset.create(listener.queue(),
                            ReadOffset.lastConsumed()))
                    .consumer(Consumer.from(listener.group(), listener.consumer()))
                    .autoAcknowledge(listener.autoAck())
                    .errorHandler(listener.errorHandler())
                    .build();
            container.register(streamReadRequest, listener);
        }
        return container;
    }

    private void createGroup(RedisTemplate redisTemplate, String queueKey, String groupName) {
        StreamOperations streamOperations = redisTemplate.opsForStream();
        // 要先创建好key 不然读取组的时候会报错的
        if (!redisTemplate.hasKey(queueKey)) {
            // 创建key
            RecordId add = streamOperations.add(queueKey, MapUtil.builder().put("key", "value").build());
            streamOperations.createGroup(queueKey, groupName);
            streamOperations.delete(queueKey, add);
            return;
        }

        StreamInfo.XInfoGroups groups = streamOperations.groups(queueKey);
        if (Objects.isNull(groups) || !groups.stream().anyMatch(b -> Objects.equals(groupName, b.groupName()))) {
            log.info("redis stream queue key:{}-groupName:{} 创建成功", queueKey, groupName);
            streamOperations.createGroup(queueKey, groupName);
        }
    }

}
